package com.jy.oms.framework.asyn.task.alarm;

import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerProcessor;
import com.jy.oms.framework.asyn.task.model.TaskConfig;
import com.jy.oms.framework.asyn.task.model.TaskExceedRetryMaxNumDto;
import com.jy.oms.framework.asyn.task.model.TaskHandlerWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 任务执行报错告警服务
 *
 * @author liuquan
 */
@Service
public class TaskErrorAlarmMessageService implements ITaskAlarmMessageService, ITaskHandlerProcessor {
    private static Logger logger = LoggerFactory.getLogger(TaskErrorAlarmMessageService.class);
    @Autowired
    private ITaskHandlerManager taskHandlerManager;

    private ConcurrentHashMap<String, AlarmBo> counterMap = new ConcurrentHashMap<>();
    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

    @Override
    public void alarm(AlarmMessage alarmMessage) {
        AlarmMessageFacade.getInstance().alarm(alarmMessage);
    }

    /**
     * 单位时间内任务报错时只告警一次；
     */
    @Override
    public <T> void alarm(IMessageConvert<T> messageConvert, T msg) {
        if (!(msg instanceof TaskExceedRetryMaxNumDto)) {
            return;
        }

        try {
            TaskExceedRetryMaxNumDto taskExceedRetryMaxNumDto = (TaskExceedRetryMaxNumDto) msg;
            String taskCode = taskExceedRetryMaxNumDto.getTaskCode();

            counterMap.computeIfAbsent(taskCode, k -> new AlarmBo(new AtomicInteger(0), new AtomicBoolean(false), 0));
            AlarmBo alarmBo = counterMap.get(taskCode);
            if (alarmBo.alarm.compareAndSet(false, true)) {
                this.alarm(messageConvert.convert(msg));
            } else {
                if (alarmBo.time == 0) {
                    alarmBo.time = System.currentTimeMillis();
                }
                alarmBo.counter.incrementAndGet();
            }
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    private void errorAlarm(String taskCode, AlarmBo alarmBo) {
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskCode);
        TaskConfig taskConfig = taskHandlerWrapper.getTaskConfig();
        AlarmMessage alarmMessage = AlarmMessage.create()
                .title("统计异步任务报错次数")
                .add("任务taskCode", taskCode)
                .add("报错次数", taskConfig.getRetryNum() > 0 ? alarmBo.counter.get() * taskConfig.getRetryNum() : alarmBo.counter.get())
                .add("累计统计时间(分钟)", (System.currentTimeMillis() - alarmBo.time) / 1000 / 60);

        this.alarm(alarmMessage);
    }

    @Override
    public void processAfterTaskHandlerInitialization() {
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    for (Map.Entry<String, AlarmBo> entry : counterMap.entrySet()) {
                        String taskCode = entry.getKey();
                        AlarmBo alarmBo = entry.getValue();

                        if (alarmBo.counter.get() > 0) {
                            if(alarmBo.counter.get() > 1){
                                errorAlarm(taskCode, alarmBo);
                            }

                            alarmBo.counter.set(0);
                            alarmBo.alarm.set(false);
                            alarmBo.time = 0;
                        }
                    }
                } catch (Throwable e) {
                    logger.error("", e);
                }
            }
        }, 1, 10, TimeUnit.MINUTES);
    }

    private static class AlarmBo {
        AtomicInteger counter;
        AtomicBoolean alarm;
        long time;

        public AlarmBo(AtomicInteger counter, AtomicBoolean alarm, long time) {
            this.counter = counter;
            this.alarm = alarm;
            this.time = time;
        }
    }
}